package com.demo.rocketmq.config;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.demo.rocketmq.config.properties.Consume;
import com.demo.rocketmq.config.properties.Retry;
import com.demo.rocketmq.config.properties.RocketMqConfigPropertites;
import com.demo.rocketmq.constants.RocketMqConstants;
import com.demo.rocketmq.entity.RetrySendMsg;
import com.demo.rocketmq.listener.ServiceSharding;
import com.demo.rocketmq.mapper.RetrySendMsgMapper;
import com.demo.rocketmq.utils.RocketMqUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.util.CollectionUtils;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 从数据库获取需要重试消息，再次发送消息，确保消费者能消费消息
 * @author cyj
 */
@Slf4j
@Configuration
@ConditionalOnProperty(prefix = RocketMqConstants.CALLBACK_PROPERTIES, name = "enable", havingValue = "true")
@AutoConfigureAfter({RocketMQAutoConfiguration.class})
public class RocketCallbackConfig implements SmartInitializingSingleton {

    private final RetrySendMsgMapper retrySendMsgMapper;

    private final RocketMqConfigPropertites rocketMqConfigPropertites;

    private final RocketMQProperties rocketMQProperties;

    private final RocketMQTemplate rocketMqTemplate;

    private final ServiceSharding serviceSharding;

    private final ScheduledExecutorService scheduler = new ScheduledThreadPoolExecutor(1, new CustomizableThreadFactory(
            "重发消息定时任务ScheduledExecutorService线程池"));

    List<DefaultMQPushConsumer> list = new ArrayList<>();

    public RocketCallbackConfig(RocketMqConfigPropertites rocketMqConfigPropertites,
                                RocketMQProperties rocketMQProperties, RocketMQTemplate rocketMqTemplate,
                                RetrySendMsgMapper retrySendMsgMapper, ServiceSharding serviceSharding) {
        log.info("RocketCallbackConfig 初始化");

        this.rocketMqConfigPropertites = rocketMqConfigPropertites;
        this.rocketMQProperties = rocketMQProperties;
        this.rocketMqTemplate = rocketMqTemplate;
        this.retrySendMsgMapper = retrySendMsgMapper;
        this.serviceSharding = serviceSharding;
    }

    @Override
    public void afterSingletonsInstantiated() {
        // 重发消息
        send();

        // 消费回调消息
        consume();
    }

    private void send() {
        Retry retry = ObjectUtils.getIfNull(rocketMqConfigPropertites.getRetry(), Retry::new);

        // 带延迟时间的调度，循环执行，固定频率
        scheduler.scheduleAtFixedRate(() -> sendMsg(retry), 1, retry.getInterval(), TimeUnit.SECONDS);
    }

    private void sendMsg(Retry retry) {
        String sql = String.format("and id %% %d = %d limit 500", serviceSharding.getServices(),
                serviceSharding.getShardingValue());
        Date date = DateUtils.addMinutes(new Date(), retry.getExpireTime() * -1);
        LambdaQueryWrapper<RetrySendMsg> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(RetrySendMsg::getFlag, 0)
                .le(RetrySendMsg::getRetryNum, retry.getRetryNum())
                .lt(RetrySendMsg::getUpdateTime, date)
                .last(sql);

        List<RetrySendMsg> retrySendMsgs = retrySendMsgMapper.selectList(wrapper);
        log.info("定时执行重发消息，重发消息数量：{}", retrySendMsgs.size());
        if (retrySendMsgs.isEmpty()) {
            return;
        }

        Map<String, List<RetrySendMsg>> map = retrySendMsgs.stream()
                .collect(Collectors.groupingBy(RetrySendMsg::getTopic));

        map.forEach(this::batchSendMsg);
    }

    /**
     * 批量发送消息
     * @param topic 主题
     * @param list 消息列表
     */
    private void batchSendMsg(String topic, List<RetrySendMsg> list) {
        Pair<String, String> pair = RocketMqUtils.splitTopic(topic);
        List<Message> messages = list.stream()
                .map(s -> {
                    Message message = new Message(pair.getLeft(), pair.getRight(),
                            s.getMsg().getBytes(StandardCharsets.UTF_8));
                    message.setKeys(s.getCallbackKey());
                    message.getProperties().put(RocketMqConstants.CALLBACK_KEY, s.getCallbackKey());
                    message.getProperties().put(RocketMqConstants.CALLBACK_TOPIC_TAG, s.getCallbackTopicTag());
                    return message;
                })
                .collect(Collectors.toList());

        List<String> ids = list.stream()
                .map(RetrySendMsg::getId)
                .collect(Collectors.toList());

        try {
            rocketMqTemplate.getProducer().send(messages);
        } catch (Exception e) {
            log.info("重试异常", e);
        } finally {
            update(ids);
        }
    }

    private void update(List<String> ids) {
        LambdaUpdateWrapper<RetrySendMsg> wrapper = new LambdaUpdateWrapper<>();
        wrapper.setSql("retry_num = retry_num + 1");
        wrapper.in(RetrySendMsg::getId, ids);
        retrySendMsgMapper.update(null, wrapper);
    }

    private void consume() {
        List<Consume> consumes = rocketMqConfigPropertites.getConsume();
        if (CollectionUtils.isEmpty(consumes)) {
            return;
        }

        try {
            for (Consume consume : consumes) {
                DefaultMQPushConsumer consumer = getDefaultMQPushConsumer(consume);
                list.add(consumer);
            }
        } catch (Exception e) {
            log.error("", e);
            throw new RuntimeException("消费回调消息客户端创建失败", e);
        }
        log.info("消费回调消息客户端创建完毕");
    }

    private DefaultMQPushConsumer getDefaultMQPushConsumer(Consume consume) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consume.getConsumerGroup(), true, StringUtils.EMPTY);
//                consumer.setInstanceName(RocketMQUtil.getInstanceName(rocketMQProperties.getNameServer()));
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        consumer.setConsumeThreadMax(64);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.subscribe(consume.getTopic(), consume.getTag());
        consumer.setMessageListener(new ConsumeCallbackMsg());
        consumer.start();
        return consumer;
    }

    private class ConsumeCallbackMsg implements MessageListenerConcurrently {

        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            List<String> list = new ArrayList<>();
            for (MessageExt messageExt : msgs) {
                log.info("callback received msg: {}", messageExt);
                try {
                    String msg = new String(messageExt.getBody(), StandardCharsets.UTF_8);
                    list.add(msg);
                } catch (Exception e) {
                    log.warn("callback consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}",
                            messageExt.getMsgId(), messageExt.getTopic(), messageExt.getReconsumeTimes(), e);
                    context.setDelayLevelWhenNextConsume(0);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }

            if (!list.isEmpty()) {
                LambdaUpdateWrapper<RetrySendMsg> wrapper = new LambdaUpdateWrapper<>();
                wrapper.set(RetrySendMsg::getFlag, 1).in(RetrySendMsg::getCallbackKey, list);
                retrySendMsgMapper.update(null, wrapper);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    }
}
